package com.example.demo.config.listener;

import com.example.demo.config.enums.ConsumerBeanEnum;
import com.example.demo.config.service.RocketMQConsumerService;
import java.nio.charset.StandardCharsets;
import java.time.LocalDateTime;
import java.util.List;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ApplicationListener;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.event.ContextRefreshedEvent;

/**
 * @author yongen
 * @description: 消费消息监听者实现类
 * @date 2022/2/9 2:56 PM
 */
@Configuration
@Slf4j
public class DefaultConsumerListener extends AbstractConsumerListener implements ApplicationContextAware,
    ApplicationListener<ContextRefreshedEvent> {

  /**
   * 上下文
   */
  private ApplicationContext applicationContext;

  @Override
  public ConsumeConcurrentlyStatus onMessage(List<MessageExt> msgs) {
    for (MessageExt msg : msgs) {
      String topicTag = msg.getTopic() + ":" + msg.getTags();
      ConsumerBeanEnum mqConsumerBeanEnum = ConsumerBeanEnum.getBeanByTopicTag(topicTag);
      if (mqConsumerBeanEnum == null) {
        break;
      }
      ConsumeConcurrentlyStatus consumerStatus = null;
      Object serviceBean = applicationContext.getBean(mqConsumerBeanEnum.getBeanName());
      if (null == serviceBean) {
        break;
      }
      if (serviceBean instanceof RocketMQConsumerService) {
        String message = new String(msg.getBody(), StandardCharsets.UTF_8);
        RocketMQConsumerService consumerService = (RocketMQConsumerService) serviceBean;
        // 预处理
        consumerService.beforeHandler(message);
        // 处理
        consumerStatus = consumerService.handle(message);
        // 处理之后
        consumerService.afterHandler(message, LocalDateTime.now(), consumerStatus);
      }
    }
    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
  }

  @Override
  public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
    this.applicationContext = applicationContext;
  }

  @Override
  public void onApplicationEvent(ContextRefreshedEvent contextRefreshedEvent) {
    try {
      // 订阅同一个topic下的多个tag
      super.listener("T_DEMO_PROJECT", "ADD_USER || ADD_ORDER");
    } catch (MQClientException e) {
      log.error("consumer error");
    }
  }
}
